Use ActiveJobs interface
Dominik Sander
6 changed files with 42 additions and 35 deletions
+ 15
- 0
app/jobs/agent_check_job.rb
@@ -0,0 +1,15 @@ |
||
| 1 |
+class AgentCheckJob < ActiveJob::Base |
|
| 2 |
+ # Given an Agent id, load the Agent, call #check on it, and then save it with an updated `last_check_at` timestamp. |
|
| 3 |
+ def perform(agent_id) |
|
| 4 |
+ agent = Agent.find(agent_id) |
|
| 5 |
+ begin |
|
| 6 |
+ return if agent.unavailable? |
|
| 7 |
+ agent.check |
|
| 8 |
+ agent.last_check_at = Time.now |
|
| 9 |
+ agent.save! |
|
| 10 |
+ rescue => e |
|
| 11 |
+ agent.error "Exception during check. #{e.message}: #{e.backtrace.join("\n")}"
|
|
| 12 |
+ raise |
|
| 13 |
+ end |
|
| 14 |
+ end |
|
| 15 |
+end |
+ 16
- 0
app/jobs/agent_receive_job.rb
@@ -0,0 +1,16 @@ |
||
| 1 |
+class AgentReceiveJob < ActiveJob::Base |
|
| 2 |
+ # Given an Agent id and an array of Event ids, load the Agent, call #receive on it with the Event objects, and then |
|
| 3 |
+ # save it with an updated `last_receive_at` timestamp. |
|
| 4 |
+ def perform(agent_id, event_ids) |
|
| 5 |
+ agent = Agent.find(agent_id) |
|
| 6 |
+ begin |
|
| 7 |
+ return if agent.unavailable? |
|
| 8 |
+ agent.receive(Event.where(:id => event_ids).order(:id)) |
|
| 9 |
+ agent.last_receive_at = Time.now |
|
| 10 |
+ agent.save! |
|
| 11 |
+ rescue => e |
|
| 12 |
+ agent.error "Exception during receive. #{e.message}: #{e.backtrace.join("\n")}"
|
|
| 13 |
+ raise |
|
| 14 |
+ end |
|
| 15 |
+ end |
|
| 16 |
+end |
+ 6
- 32
app/models/agent.rb
@@ -387,24 +387,11 @@ class Agent < ActiveRecord::Base |
||
| 387 | 387 |
end |
| 388 | 388 |
end |
| 389 | 389 |
|
| 390 |
- # Given an Agent id and an array of Event ids, load the Agent, call #receive on it with the Event objects, and then |
|
| 391 |
- # save it with an updated `last_receive_at` timestamp. |
|
| 392 |
- # |
|
| 393 |
- # This method is tagged with `handle_asynchronously` and will be delayed and run with delayed_job. It accepts Agent |
|
| 394 |
- # and Event ids instead of a literal ActiveRecord models because it is preferable to serialize delayed_jobs with ids. |
|
| 390 |
+ # This method will enqueue an AgentReceiveJob job. It accepts Agent and Event ids instead of a literal ActiveRecord |
|
| 391 |
+ # models because it is preferable to serialize jobs with ids. |
|
| 395 | 392 |
def async_receive(agent_id, event_ids) |
| 396 |
- agent = Agent.find(agent_id) |
|
| 397 |
- begin |
|
| 398 |
- return if agent.unavailable? |
|
| 399 |
- agent.receive(Event.where(:id => event_ids).order(:id)) |
|
| 400 |
- agent.last_receive_at = Time.now |
|
| 401 |
- agent.save! |
|
| 402 |
- rescue => e |
|
| 403 |
- agent.error "Exception during receive. #{e.message}: #{e.backtrace.join("\n")}"
|
|
| 404 |
- raise |
|
| 405 |
- end |
|
| 393 |
+ AgentReceiveJob.perform_later(agent_id, event_ids) |
|
| 406 | 394 |
end |
| 407 |
- handle_asynchronously :async_receive |
|
| 408 | 395 |
|
| 409 | 396 |
# Given a schedule name, run `check` via `bulk_check` on all Agents with that schedule. |
| 410 | 397 |
# This is called by bin/schedule.rb for each schedule in `SCHEDULES`. |
@@ -425,24 +412,11 @@ class Agent < ActiveRecord::Base |
||
| 425 | 412 |
end |
| 426 | 413 |
end |
| 427 | 414 |
|
| 428 |
- # Given an Agent id, load the Agent, call #check on it, and then save it with an updated `last_check_at` timestamp. |
|
| 429 |
- # |
|
| 430 |
- # This method is tagged with `handle_asynchronously` and will be delayed and run with delayed_job. It accepts an Agent |
|
| 431 |
- # id instead of a literal Agent because it is preferable to serialize delayed_jobs with ids, instead of with the full |
|
| 432 |
- # Agents. |
|
| 415 |
+ # This method will enqueue an AgentCheckJob job. It accepts an Agent id instead of a literal Agent because it is |
|
| 416 |
+ # preferable to serialize job with ids, instead of with the full Agents. |
|
| 433 | 417 |
def async_check(agent_id) |
| 434 |
- agent = Agent.find(agent_id) |
|
| 435 |
- begin |
|
| 436 |
- return if agent.unavailable? |
|
| 437 |
- agent.check |
|
| 438 |
- agent.last_check_at = Time.now |
|
| 439 |
- agent.save! |
|
| 440 |
- rescue => e |
|
| 441 |
- agent.error "Exception during check. #{e.message}: #{e.backtrace.join("\n")}"
|
|
| 442 |
- raise |
|
| 443 |
- end |
|
| 418 |
+ AgentCheckJob.perform_later(agent_id) |
|
| 444 | 419 |
end |
| 445 |
- handle_asynchronously :async_check |
|
| 446 | 420 |
end |
| 447 | 421 |
end |
| 448 | 422 |
|
+ 1
- 1
app/models/agents/email_agent.rb
@@ -35,7 +35,7 @@ module Agents |
||
| 35 | 35 |
incoming_events.each do |event| |
| 36 | 36 |
log "Sending digest mail to #{user.email} with event #{event.id}"
|
| 37 | 37 |
recipients(event.payload).each do |recipient| |
| 38 |
- SystemMailer.delay.send_message(:to => recipient, :subject => interpolated(event)['subject'], :headline => interpolated(event)['headline'], :body => interpolated(event)['body'], :groups => [present(event.payload)]) |
|
| 38 |
+ SystemMailer.send_message(:to => recipient, :subject => interpolated(event)['subject'], :headline => interpolated(event)['headline'], :body => interpolated(event)['body'], :groups => [present(event.payload)]).deliver_later |
|
| 39 | 39 |
end |
| 40 | 40 |
end |
| 41 | 41 |
end |
+ 1
- 1
app/models/agents/email_digest_agent.rb
@@ -42,7 +42,7 @@ module Agents |
||
| 42 | 42 |
groups = self.memory['queue'].map { |payload| present(payload) }
|
| 43 | 43 |
log "Sending digest mail to #{user.email} with events [#{ids}]"
|
| 44 | 44 |
recipients.each do |recipient| |
| 45 |
- SystemMailer.delay.send_message(:to => recipient, :subject => interpolated['subject'], :headline => interpolated['headline'], :groups => groups) |
|
| 45 |
+ SystemMailer.send_message(:to => recipient, :subject => interpolated['subject'], :headline => interpolated['headline'], :groups => groups).deliver_later |
|
| 46 | 46 |
end |
| 47 | 47 |
self.memory['queue'] = [] |
| 48 | 48 |
self.memory['events'] = [] |
+ 3
- 1
config/application.rb
@@ -13,7 +13,7 @@ module Huginn |
||
| 13 | 13 |
# -- all .rb files in that directory are automatically loaded. |
| 14 | 14 |
|
| 15 | 15 |
# Custom directories with classes and modules you want to be autoloadable. |
| 16 |
- config.autoload_paths += %W(#{config.root}/lib #{config.root}/app/presenters)
|
|
| 16 |
+ config.autoload_paths += %W(#{config.root}/lib #{config.root}/app/presenters #{config.root}/app/jobs)
|
|
| 17 | 17 |
|
| 18 | 18 |
# Activate observers that should always be running. |
| 19 | 19 |
# config.active_record.observers = :cacher, :garbage_collector, :forum_observer |
@@ -52,5 +52,7 @@ module Huginn |
||
| 52 | 52 |
|
| 53 | 53 |
# Do not swallow errors in after_commit/after_rollback callbacks. |
| 54 | 54 |
config.active_record.raise_in_transactional_callbacks = true |
| 55 |
+ |
|
| 56 |
+ config.active_job.queue_adapter = :delayed_job |
|
| 55 | 57 |
end |
| 56 | 58 |
end |